package com.tqz.rc.client.push;

import com.tqz.rc.api.UdpPushAckPacket;
import com.tqz.rc.client.naming.cache.ServiceInfoHolder;
import com.tqz.rc.common.constant.Constants;
import com.tqz.rc.common.util.DateUtil;
import com.tqz.rc.common.util.IoUtils;
import com.tqz.rc.common.util.JacksonUtils;
import com.tqz.rc.common.util.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.SocketException;
import java.util.Properties;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;

/**
 * <p> {@link PushReceiver} 用于接收Nacos服务端的推送，
 * 初始化时会创建DatagramSocket使用UDP的方式接收推送。
 * 会启动1个名为com.alibaba.nacos.naming.push.receiver的线程。
 *
 * @author tianqingzhao
 * @since 2021/9/9 15:15
 */
public class PushReceiver implements Runnable {

    private static final Logger log = LoggerFactory.getLogger(PushReceiver.class);

    private DatagramSocket udpSocket;

    private ScheduledExecutorService executorService;

    private ServiceInfoHolder serviceInfoHolder;

    private static final String PUSH_PACKAGE_TYPE_DOM = "dom";

    private static final String PUSH_PACKAGE_TYPE_SERVICE = "service";

    private static final String PUSH_PACKAGE_TYPE_DUMP = "dump";

    public PushReceiver(ServiceInfoHolder serviceInfoHolder, Properties properties) {
        try {
            this.serviceInfoHolder = serviceInfoHolder;
            String udpPort = properties.getProperty("udp.port");
            this.udpSocket = new DatagramSocket(Integer.valueOf(udpPort));
            this.executorService = new ScheduledThreadPoolExecutor(1, new ThreadFactory() {
                @Override
                public Thread newThread(Runnable r) {
                    Thread thread = new Thread(r);
                    thread.setDaemon(true);
                    thread.setName("com.tqz.rc.client.push.receiver");
                    return thread;
                }
            });

            executorService.execute(this);
        } catch (SocketException e) {
            log.error("[NA] init udp socket failed", e);
        }
    }

    @Override
    public void run() {
        log.info("客户端等待接收服务端服务变更的推送，{}", DateUtil.getCurDateTime());
        try {
            byte[] buffer = new byte[Constants.UDP_MSS];
            DatagramPacket packet = new DatagramPacket(buffer, buffer.length);
            udpSocket.receive(packet);
            log.info("客户端已接收到服务端服务变更的推送，{}", DateUtil.getCurDateTime());

            String json = new String(IoUtils.tryDecompress(packet.getData()), Constants.CHARSET_UTF_8).trim();
            UdpPushAckPacket pushPacket = JacksonUtils.toObj(json, UdpPushAckPacket.class);
            System.out.println(pushPacket);

            String ack = processAck(pushPacket);
            // 向服务端发送处理完服务变更的响应
            udpSocket.send(new DatagramPacket(ack.getBytes(Constants.CHARSET_UTF_8), ack.getBytes(Constants.CHARSET_UTF_8).length, packet.getSocketAddress()));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public String processAck(UdpPushAckPacket pushPacket) {
        String ack;
        if (PUSH_PACKAGE_TYPE_DOM.equals(pushPacket.type) || PUSH_PACKAGE_TYPE_SERVICE.equals(pushPacket.type)) {
            // 处理服务信息
            serviceInfoHolder.processServiceInfo(pushPacket.data);

            // 发送ack给服务端
            ack = "{\"type\": \"push-ack\"" + ", \"lastRefTime\":\"" + pushPacket.lastRefTime + "\", \"data\":" + "\"\"}";
        } else if (PUSH_PACKAGE_TYPE_DUMP.equals(pushPacket.type)) {
            // dump data to server 此版本可忽略
            ack = "{\"type\": \"dump-ack\"" + ", \"lastRefTime\": \"" + pushPacket.lastRefTime + "\", \"data\":"
                    + "\"" + StringUtils.escapeJavaScript(JacksonUtils.toJson(serviceInfoHolder.getServiceInfoMap())) + "\"}";
        } else {
            // 什么也不做，仅发送确认机制
            ack = "{\"type\": \"unknown-ack\"" + ", \"lastRefTime\":\"" + pushPacket.lastRefTime + "\", \"data\":" + "\"\"}";
        }

        return ack;
    }

}
